package com.shujia.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo20PageRank {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
      .setAppName("pi")
      .setMaster("local")

    val sc = new SparkContext(conf)


    //读取数据
    val pageRDD: RDD[String] = sc.textFile("spark/data/pagerank.txt")


    //多多次使用的rdd进行缓存
    pageRDD.cache()


    //网页数量
    val num: Long = pageRDD.count()


    //网页出链列表
    val clRDD: RDD[(String, List[String])] = pageRDD.map(line => {
      val split: Array[String] = line.split("\\|")
      //当前网页
      val page: String = split(0)

      //当前网页出链列表
      val cl: List[String] = split(1).split(",").toList
      (page, cl)
    })


    //2 给每个网页一个初始值
    var initPRRDD: RDD[PageRank] = pageRDD.map(line => {
      val split: Array[String] = line.split("\\|")
      //当前网页
      val page: String = split(0)

      //当前网页出链列表
      val cl: List[String] = split(1).split(",").toList

      //初始pr为1
      PageRank(page, cl, 1.0)
    })


    var flag: Boolean = true


    while (flag) {

      //2 将每一个页面的pr值平分给出链列表
      val avgPRRDD: RDD[(String, Double)] = initPRRDD.flatMap(p => {

        //出链列表
        val cl: List[String] = p.cl

        //当前网页pr值
        val pr: Double = p.pr

        //分给出链的pr值
        val avgPr: Double = pr / cl.length

        //加上pr值
        val tuples: List[(String, Double)] = cl.map(page => (page, avgPr))

        //返回,由于外面有的flatmap .会被压扁
        tuples
      })

      //网页新的pr值
      val currPRRDD: RDD[(String, Double)] = avgPRRDD.reduceByKey((x, y) => x + y)

      //将出链列表加上
      val joinRDD: RDD[(String, (Double, List[String]))] = currPRRDD.join(clRDD)

      //整理数据
      val resultRDD: RDD[PageRank] = joinRDD.map {
        case (page, (pr, cl)) => {
          PageRank(page, cl, pr)
        }
      }


      //计算所有网页当前pr值和上一次pr值的差值平均值

      val initKVRDD: RDD[(String, Double)] = initPRRDD.map(p => (p.page, p.pr))


      val currKVRDD: RDD[(String, Double)] = resultRDD.map(p => (p.page, p.pr))


      val prJoinRDD: RDD[(String, (Double, Double))] = currKVRDD.join(initKVRDD)

      //计算差值
      val chaRDD: RDD[Double] = prJoinRDD.map {
        case (page, (currPr, initPr)) => {
          //计算差值,取整
          Math.abs(currPr - initPr)
        }
      }

      //计算差值平均值
      val avg: Double = chaRDD.sum() / num


      if (avg < 0.01) {
        flag = false
      }


      //第二次迭代使用最新的pr值
      initPRRDD = resultRDD


      initPRRDD.foreach(println)

    }


  }

  case class PageRank(page: String, cl: List[String], pr: Double)

}
